#ifndef __M_MESSAGE_H__
#define __M_MESSAGE_H__
#include "../mqcommon/logger.hpp"
#include "../mqcommon/filer.hpp"
#include "../mqcommon/uuider.hpp"
#include "../mqcommon/message.pb.h"

#include <iostream>
#include <unordered_map>
#include <string>
#include <mutex>
#include <memory>
#include <list>
#include <sstream>

#define DATAFILE_SUBFIX ".mqd"
#define TMPFILE_SUBFIX ".mqd.tmp"
#define VAILD "1"   // 有效数据 持久化存储
#define UNVAILD "0" // 无效数据 非持久化
namespace zhong
{
    // Message的具体实现 我们已经在proto中实现
    // struct BasicProperties
    // {
    //     std::string id;
    //     DeliveryMode delivery_mode;
    //     std::string routing_key;
    // };
    // struct Payload
    // {
    //     struct BasicProperties properties;
    //     std::string body;
    //     std::string valid;
    // };
    // struct Message
    // {
    //     struct Payload payload;
    //     size_t offset;
    //     size_t length;
    // };

    using MessagePtr = std::shared_ptr<zhong::Message>; // message.pb.h中的类

    /*
        消息的持久化、内存级别的管理是以队列为单位的！！！一个队列里面存放若干个message
        并且与交换机管理、队列管理、绑定管理存储于数据库不同，队列中消息管理是存储在磁盘文件中
    */

    class MessageMapper
    {
        friend class QueueMessage;

    public:
        MessageMapper(const std::string &data_dir, const std::string &queue_name)
            : _queue_name(queue_name), _data_file(data_dir)
        {
            if (_data_file.back() != '/')
            {
                _data_file.push_back('/');
            }
            // data_base_dir/
            _tmp_file = _data_file + _queue_name + TMPFILE_SUBFIX;
            _data_file = _data_file + _queue_name + DATAFILE_SUBFIX; // ?????
                                                                     // 创建目录
            std::string path = Filer::parentDirectory(_data_file);
            // 文件夹创建时，如果存在重复创建会出问题
            if (Filer(path).exist() == false)
            {
                assert(Filer::createDirectory(path));
            }
            // 创建消息文件
            createMessageFile();
        }
        bool createMessageFile()
        {
            // 如果文件存在，重复创建，会覆盖原有文件，导致获得空数据
            if (Filer(_data_file).exist() == true)
            {
                return true;
            }
            if (Filer::createFile(_data_file) == false)
            {
                ERR_LOG("创建文件[%s]失败，原因[%s]", _data_file.c_str(), strerror(errno));
                return false;
            }
            return true;
        }
        // 删除数据文件
        void removeMessageFile()
        {
            Filer::removeFile(_data_file);
            Filer::removeFile(_tmp_file);
        }
        // 添加数据到数据文件中
        bool insert(MessagePtr &message)
        {
            return insertDiv(message, _data_file);
        }
        // 重置消息的标志位
        bool remove(MessagePtr &message) // 本质上是重置有效标志位，等待后续gc的垃圾回收
        {
            // 有效标志位重置 有效：'1' 无效："0"，用来gc垃圾回收判断 是否可以省略？而是直接通过消息的持久化标志来进行！
            message->mutable_payload()->set_valid(UNVAILD);
            std::string payload = message->payload().SerializeAsString();
            if (payload.size() != message->length()) // 判断有效载荷是否大小
            {
                ERR_LOG("修改前后数据长度不一致，无法进行写入");
                return false;
            }
            // 将有效载和覆盖回原先数据文件中
            Filer filer(_data_file);
            // 写入的内容：有效载荷，起始位置：此时的offset位置，大小：有效载荷的大小
            if (filer.write(payload.c_str(), message->offset(), payload.size()) == false)
            {
                ERR_LOG("向队列数据文件[%s]覆盖数据失败，原因[%s]", _data_file.c_str(), strerror(errno));
                return false;
            }
            return true;
        }
        // 垃圾回收模块
        std::list<MessagePtr> gc() // 本质上是将无效消息从data_file中除去
        {
            std::list<zhong::MessagePtr> message_list;
            if (loadMessageList(message_list) == false)
            {
                ERR_LOG("从loadMessageList()加载有效数据失败");
                return message_list;
            }

            // DEBUG_LOG("垃圾回收，加载的有效数据个数: %d", message_list.size());
            // 2.将有效数据，序列化存储到临时文件中
            Filer::createFile(_tmp_file);
            for (auto &message : message_list)
            {
                // DEBUG_LOG("向临时文件内写入数据为[%s]，大小: %d", message->payload().body().c_str(), message->payload().body().length());
                if (insertDiv(message, _tmp_file) == false)
                {
                    ERR_LOG("向临时文件写入有效数据失败，原因[%s]", strerror(errno));
                    return message_list;
                }
            }
            // DEBUG_LOG("垃圾回收，向临时文件写入数据完毕，文件大小: %ld", Filer(_tmp_file).size());

            // 3.删除源文件
            if (Filer::removeFile(_data_file) == false)
            {
                ERR_LOG("删除源文件失败，原因[%s]", strerror(errno));
                return message_list;
            }
            // 4.修改临时文件名
            if (Filer(_tmp_file).rename(_data_file) != false) // 创建临时对象Filer(_tmp_file)来修改名字为_data_file
            {
                // 注意rename成功返回值为0(int)，而0(bool)为false，所以Filer(_tmp_file).rename(_data_file) == false表示rename成功
                ERR_LOG("修改临时文件名称失败，原因[%s]", strerror(errno));
                return message_list;
            }
            // 5.返回新的有效数据
            return message_list;
        }

    private:
        // 提供给数据文件和临时文件的insert接口
        bool insertDiv(MessagePtr &message, const std::string &file_name)
        {
            // 将网络中获取的有效载荷转为字符流
            std::string payload = message->payload().SerializeAsString();
            // 将有效载和写入文件中
            Filer filer(file_name);
            size_t file_size = filer.size();
            // 先写入sizeof(size_t)字节长度
            size_t message_size = payload.size();
            if (filer.write((char *)&message_size, file_size, sizeof(size_t)) == false)
            {
                ERR_LOG("向队列数据文件[%s]写入4字节长度失败，原因[%s]", _data_file.c_str(), strerror(errno));
                return false;
            }
            // 在写入数据内容
            if (filer.write(payload.c_str(), file_size + sizeof(size_t), payload.size()) == false)
            {
                ERR_LOG("向队列数据文件[%s]写入数据失败，原因[%s]", _data_file.c_str(), strerror(errno));
                return false;
            }
            // 更新数据文件中message对应的属性信息
            message->set_offset(file_size + sizeof(size_t)); // 起始位置：之前未插入前文件的末尾
            message->set_length(payload.size());             // 消息的长度：有效载荷的大小
            return true;
        }
        // 加载源文件中的所有有效数据
        bool loadMessageList(std::list<zhong::MessagePtr> &message_list)
        {
            Filer data_filer(_data_file);
            size_t file_size = data_filer.size();
            size_t offset = 0, payload_size = 0; // ??? 为什么91行这样子，用类型强转，类似stoi
            // 当offset小于文件内容大小，表示还存在有效内容
            while (offset < file_size)
            {
                // 读取有效载荷长度
                if (data_filer.read((char *)&payload_size, offset, sizeof(size_t)) == false)
                {
                    ERR_LOG("读取有效载荷长度失败，原因[%s]", strerror(errno));
                    return false;
                }
                // 文件内容形式  |4字节的长度|大小未知的Payload|4字节的长度|大小未知的Payload|
                offset += sizeof(size_t);
                // 读取有效载荷
                std::string payload(payload_size, '\0'); // 初始化payload并占位
                if (data_filer.read(&payload[0], offset, payload_size) == false)
                {
                    ERR_LOG("读取有效载荷失败，原因[%s]", strerror(errno));
                    return false;
                }
                offset += payload_size;
                // 初始化消息对象
                MessagePtr message = std::make_shared<Message>();
                message->mutable_payload()->ParseFromString(payload); // 对有效载荷进行反序列化
                // 判断是否为有效数据
                if (message->payload().valid() == UNVAILD)
                {
                    INFO_LOG("加载到无效消息[%s]", message->payload().body().c_str());
                    continue; // 无效消息不需要保存在message_list
                }
                message_list.push_back(message);
            }
            return true;
        }

    private:
        std::string _queue_name;
        std::string _data_file;
        std::string _tmp_file;
    };

    /*
        注意这里要区分当前QueueMessage 和 queue.hpp中的MessageQueue
        QueueMessage -> 在内存级别的消息队列的消息管理
        MessageQueue -> 消息队列的队列的具体属性
    */

    // 内存级别的消息管理的队列单元
    class QueueMessage
    {
        friend class MessageManager;

    public:
        using ptr = std::shared_ptr<QueueMessage>;

        QueueMessage(const std::string &data_dir, const std::string &queue_name)
            : _mapper(data_dir, queue_name), _valid_count(0), _total_count(0), _queue_name(queue_name)
        {
        }
        // 加载获取队列中的有效消息
        void getAllQueueMessage()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            // 通过gc直接恢复有效消息
            _message_list = _mapper.gc();

            for (auto &message : _message_list)
            {
                _durable_map.insert(std::make_pair(message->payload().properties().id(), message));
            }
            _valid_count = _total_count = _durable_map.size(); // _message_list.size()
        }
        // 实现消息存储
        bool insert(const BasicProperties *bp, const std::string &body, bool queue_durable) // 传入消息的属性和消息内容，投递模式
        {
            
            zhong::DeliveryMode mode;
            // 1.构造消息对象
            MessagePtr message = std::make_shared<Message>();
            message->mutable_payload()->set_body(body);
            // 设置基础属性结构体
            if (bp != nullptr)
            {
                message->mutable_payload()->mutable_properties()->set_id(bp->id());
                message->mutable_payload()->mutable_properties()->set_routing_key(bp->routing_key());
                // 当bp不为空时，现在同时有了queue_durable和bp->delivery_mode()
                // 如果队列为持久化，那么消息的持久化就根据消息内部决定
                mode = queue_durable == true ? bp->delivery_mode() : zhong::DeliveryMode::UNDURABLE;
                message->mutable_payload()->mutable_properties()->set_delivery_mode(mode);
            }
            else // bp为空时，需要开发者自行生成
            {
                message->mutable_payload()->mutable_properties()->set_id(UUIDer::uuid()); // 通过uuid来实现随机化id
                message->mutable_payload()->mutable_properties()->set_routing_key("");
                // 如果bp不存在，那么队列为持久化，这时消息也为持久化
                mode = queue_durable == true ? zhong::DeliveryMode::DURABLE : zhong::DeliveryMode::UNDURABLE;
                message->mutable_payload()->mutable_properties()->set_delivery_mode(mode);
            }
            // 对公共部分容器进行加锁
            std::unique_lock<std::mutex> lock(_mutex);
            // 2.判断消息是否需要持久化
            if (message->payload().properties().delivery_mode() == DeliveryMode::DURABLE) // DeliveryMode::DURABLE表示持久化
            {
                // 设置标准位
                message->mutable_payload()->set_valid(VAILD);
                // 进行持久化存储
                if (_mapper.insert(message) == false)
                {
                    ERR_LOG("持久化存储[%s]信息失败，原因[%s]", body.c_str(), strerror(errno));
                    return false;
                }
                // 添加进持久化哈希表
                _durable_map.insert(std::make_pair(message->payload().properties().id(), message));
                // 计数++
                _valid_count++;
                _total_count++;
            }
            // 内存级别的数据不用设置过多的消息属性？？？proto实现的类会提供调用的接口
            // message的原生结构只是为了文件操作时找到offset起始位置，length消息长度，payload中的body信息内容？

            // 3.进行内存消息管理
            _message_list.push_back(message);
            return true;
        }
        // 提供待推送消息list的头消息
        MessagePtr front()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            if (_message_list.size() == 0)
            {
                return MessagePtr();
            }
            MessagePtr message = _message_list.front();
            // 待推送消息推送后，需要添加到待确认模块中
            _wait_ack_map.insert(std::make_pair(message->payload().properties().id(), message));
            _message_list.pop_front();
            return message;
        }
        // 通过消息名删除消息
        bool remove(const std::string &message_id)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            // 1.从待确认map中查找消息
            auto it = _wait_ack_map.find(message_id);
            if (it == _wait_ack_map.end())
            {
                DEBUG_LOG("没有找到消息[%s]", message_id.c_str());
                return true;
            }
            // 2.消息是否持久化，是否删除
            if (it->second->payload().properties().delivery_mode() == DeliveryMode::DURABLE)
            {
                // 3.删除持久化消息
                _mapper.remove(it->second);
                _durable_map.erase(message_id);
                // 计数减减
                _valid_count--;
                // 进行垃圾回收
                gc();
            }
            // 4.删除内存中消息
            _wait_ack_map.erase(message_id);
            // 会不会是外部调用front获得一个信息后，在通过这个信息的id来在内存中进行确认，接着删除，从待确认map中除去
            return true;
        }
        // 待推送信息数目
        size_t valid_size()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            // MessageMapper中我们就定义std::list<MessagePtr>里面存放的就是有效消息
            return _message_list.size();
        }
        // 总的持久化消息数目
        size_t total_size()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            return _total_count;
        }
        // 有效持久化消息数目
        size_t durable_count()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            return _durable_map.size();
        }
        // 待确认消息数目
        size_t wait_ack_count()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            return _wait_ack_map.size();
        }
        void clear()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _mapper.removeMessageFile();
            _message_list.clear();
            _durable_map.clear();
            _wait_ack_map.clear();
            _valid_count = 0;
            _total_count = 0;
        }

    private:
        bool GCCheck()
        {
            // 持久化消息总量大于200，并且有效比例低于50%
            if (_total_count >= 2000 && _valid_count * 10 / _total_count < 5)
            {
                return true;
            }
            return false;
        }
        // 垃圾回收并更新类内成员
        void gc()
        {
            if (GCCheck() == false) // 不需要进行垃圾回收，直接退出gc
                return;
            // 1.进行垃圾回收，获得有效信息链表
            std::list<MessagePtr> message_list = _mapper.gc();
            for (auto &message : message_list)
            {

                // 2.更新消息在内存中的实际存储位置
                auto it = _durable_map.find(message->payload().properties().id());
                // 一般不会存在message_list中包含的message个体，map中没有 --- 存疑是否需要判断
                if (it == _durable_map.end())
                {
                    ERR_LOG("垃圾回收后，有一条持久化消息，在内存中没有进行管理!");
                    _message_list.push_back(message);
                    _durable_map.insert(std::make_pair(message->payload().properties().id(), message));
                    continue;
                }
                it->second->set_offset(message->offset());
                it->second->set_length(message->length());
            }
            // 更新当前有效信息数量 和 总的持久化消息数量
            _valid_count = _total_count = message_list.size();
        }

    private:
        std::mutex _mutex;
        std::string _queue_name;
        size_t _valid_count; // 有效持久化消息量 --- 持久化的消息了
        size_t _total_count; // 总的持久化消息量
        MessageMapper _mapper;
        std::list<MessagePtr> _message_list;                       // 待推送消息list
        std::unordered_map<std::string, MessagePtr> _durable_map;  // 持久化消息map
        std::unordered_map<std::string, MessagePtr> _wait_ack_map; // 待确认消息map
    };

    // 提供给外部进行消息处理的类
    class MessageManager
    {
    public:
        using ptr = std::shared_ptr<MessageManager>;
        MessageManager(const std::string &data_dir) : _data_dir(data_dir) {}
        // 初始化队列消息的句柄
        void initQueueMessage(const std::string &queue_name)
        {
            QueueMessage::ptr queue_message;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_messages.find(queue_name);
                if (it != _queue_messages.end())
                {
                    DEBUG_LOG("队列[%s]已存在，不需要再次初始化", queue_name.c_str());
                    return;
                }

                queue_message = std::make_shared<QueueMessage>(_data_dir, queue_name);
                _queue_messages.insert(std::make_pair(queue_name, queue_message));
            }
            // 不在当前锁模块进行恢复队列的消息数据
            queue_message->getAllQueueMessage();
        }
        // 删除队列消息的句柄
        void destoryQueueMessage(const std::string &queue_name)
        {
            QueueMessage::ptr queue_message;
            // 进行临界区资源释放
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_messages.find(queue_name);
                if (it == _queue_messages.end())
                {
                    ERR_LOG("队列[%s]不存在，删除非法", queue_name.c_str());
                    return;
                }
                // 在内存中清理这个队列
                queue_message = it->second;
                _queue_messages.erase(queue_name);
            }
            // 用这个智能指针来释放这个队列
            queue_message->clear();
            // 另外这个队列在clear时，自己内部有锁，不需要参与公共资源_mutex的锁竞争
            // 并且释放这个队列耗时较长，也不适合用_mutex锁起来
        }
        // 添加一个消息数据到队列中
        bool insert(const std::string &queue_name, const BasicProperties *bp, const std::string &body, bool queue_durable)
        {
            QueueMessage::ptr queue_message;
            // 进行临界区资源释放
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_messages.find(queue_name);
                if (it == _queue_messages.end())
                {
                    ERR_LOG("队列[%s]不存在，无法插入新的消息数据", queue_name.c_str());
                    return false;
                }
                // 通过_queue_messages的迭代器获取到这个队列的指针
                queue_message = it->second;
            }
            return queue_message->insert(bp, body, queue_durable);
        }
        // 获取队列消息的队头消息
        MessagePtr front(const std::string &queue_name)
        {
            QueueMessage::ptr queue_message;
            // 进行临界区资源释放
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_messages.find(queue_name);
                if (it == _queue_messages.end())
                {
                    ERR_LOG("队列[%s]不存在，无法获取队头数据", queue_name.c_str());
                    return MessagePtr();
                }
                // 通过_queue_messages的迭代器获取到这个队列的指针
                queue_message = it->second;
            }
            return queue_message->front();
        }
        void ack(const std::string &queue_name, const std::string &message_id)
        {
            QueueMessage::ptr queue_message;
            // 进行临界区资源释放
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_messages.find(queue_name);
                if (it == _queue_messages.end())
                {
                    ERR_LOG("队列[%s]不存在，无法进行消息[%s]确认", queue_name.c_str(), message_id.c_str());
                    return;
                }
                // 通过_queue_messages的迭代器获取到这个队列的指针
                queue_message = it->second;
            }
            // 确认的本质就是删除
            queue_message->remove(message_id);
            return;
        }
        // 获取待推送消息数目
        size_t valid_size(const std::string &queue_name)
        {
            QueueMessage::ptr queue_message;
            // 进行临界区资源释放
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_messages.find(queue_name);
                if (it == _queue_messages.end())
                {
                    ERR_LOG("队列[%s]不存在，获取待推送消息数目失败", queue_name.c_str());
                    return false;
                }
                // 通过_queue_messages的迭代器获取到这个队列的指针
                queue_message = it->second;
            }
            return queue_message->valid_size();
        }
        // 获取总的持久化消息数目
        size_t total_size(const std::string &queue_name)
        {
            QueueMessage::ptr queue_message;
            // 进行临界区资源释放
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_messages.find(queue_name);
                if (it == _queue_messages.end())
                {
                    ERR_LOG("队列[%s]不存在，获取总的持久化消息数目失败", queue_name.c_str());
                    return false;
                }
                // 通过_queue_messages的迭代器获取到这个队列的指针
                queue_message = it->second;
            }
            return queue_message->total_size();
        }
        // 获取有效持久化消息数目
        size_t durable_count(const std::string &queue_name)
        {
            QueueMessage::ptr queue_message;
            // 进行临界区资源释放
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_messages.find(queue_name);
                if (it == _queue_messages.end())
                {
                    ERR_LOG("队列[%s]不存在，获取有效持久化消息数目失败", queue_name.c_str());
                    return false;
                }
                // 通过_queue_messages的迭代器获取到这个队列的指针
                queue_message = it->second;
            }
            return queue_message->durable_count();
        }
        // 获取待确认消息数目
        size_t wait_ack_count(const std::string &queue_name)
        {
            QueueMessage::ptr queue_message;
            // 进行临界区资源释放
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_messages.find(queue_name);
                if (it == _queue_messages.end())
                {
                    ERR_LOG("队列[%s]不存在，获取待确认消息数目失败", queue_name.c_str());
                    return false;
                }
                // 通过_queue_messages的迭代器获取到这个队列的指针
                queue_message = it->second;
            }
            return queue_message->wait_ack_count();
        }
        void clear()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            for (auto &queue_message : _queue_messages)
            {
                queue_message.second->clear();
            }
        }

    private:
        std::mutex _mutex;
        std::string _data_dir;
        std::unordered_map<std::string, QueueMessage::ptr> _queue_messages;
    };
}
#endif
